package com.viknix.threadpool.manager.config.service.impl;

import com.viknix.threadpool.manager.cluster.core.Member;
import com.viknix.threadpool.manager.cluster.core.ServerMemberManager;
import com.viknix.threadpool.manager.cluster.util.AuthHeaderUtil;
import com.viknix.threadpool.manager.config.constant.Constants;
import com.viknix.threadpool.manager.config.executor.ConfigExecutor;
import com.viknix.threadpool.manager.config.notify.NotifyTask;
import com.viknix.threadpool.manager.http.Callback;
import com.viknix.threadpool.manager.http.Header;
import com.viknix.threadpool.manager.http.TpmAsyncRestTemplate;
import com.viknix.threadpool.manager.http.Query;
import com.viknix.threadpool.manager.http.factory.HttpClientManager;
import com.viknix.threadpool.manager.server.dependence.monitor.NotifyCenter;
import com.viknix.threadpool.manager.server.dependence.monitor.event.ConfigDataChangeEvent;
import com.viknix.threadpool.manager.server.dependence.monitor.event.Event;
import com.viknix.threadpool.manager.server.dependence.monitor.listeners.Listener;
import com.viknix.threadpool.manager.server.dependence.result.RestResult;
import com.viknix.threadpool.manager.server.dependence.util.EnvUtil;
import com.viknix.threadpool.manager.server.dependence.util.InetUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.UnsupportedEncodingException;
import java.net.URLEncoder;
import java.text.MessageFormat;
import java.util.Collection;
import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.TimeUnit;

/**
 * @Author: Dongqi
 * @Date: 2021/11/18 17:38
 * @Version 1.0
 * @Description: 异步通知服务。
 */
@Slf4j
@Service
public class AsyncNotifyService {

    /**
     * In order to facilitate the system beta, without changing the notify.do interface, the new lastModifed parameter
     * is passed through the Http header.
     */
    public static final String NOTIFY_HEADER_LAST_MODIFIED = "lastModified";

    public static final String NOTIFY_HEADER_OP_HANDLE_IP = "opHandleIp";


    public static final String PERSISTENCE_EVENT_PUB = "pub";

    public static final String PERSISTENCE_EVENT_REMOVE = "remove";

    public static final String PERSISTENCE_EVENT_MERGE = "merge";

    public static final String NOTIFY_EVENT_OK = "ok";

    public static final String NOTIFY_EVENT_ERROR = "error";

    public static final String NOTIFY_EVENT_UNHEALTH = "unhealth";

    public static final String NOTIFY_EVENT_EXCEPTION = "exception";

    public static final String DUMP_EVENT_OK = "ok";

    public static final String DUMP_EVENT_REMOVE_OK = "remove-ok";

    public static final String DUMP_EVENT_ERROR = "error";

    public static final String PULL_EVENT_OK = "ok";

    public static final String PULL_EVENT_NOTFOUND = "not-found";

    public static final String PULL_EVENT_CONFLICT = "conflict";

    public static final String PULL_EVENT_ERROR = "error";

    private final TpmAsyncRestTemplate tpmAsyncRestTemplate = HttpClientManager.getTpmAsyncRestTemplate();

    private ServerMemberManager memberManager;

    @Autowired
    public AsyncNotifyService(ServerMemberManager memberManager) {
        this.memberManager = memberManager;

        // Register ConfigDataChangeEvent to NotifyCenter.
        NotifyCenter.registerToPublisher(ConfigDataChangeEvent.class, NotifyCenter.ringBufferSize);

        // Register A Subscriber to subscribe ConfigDataChangeEvent.
        NotifyCenter.registerSubscriber(new Listener() {

            @Override
            public void onEvent(Event event) {
                // Generate ConfigDataChangeEvent concurrently
                if (event instanceof ConfigDataChangeEvent) {
                    ConfigDataChangeEvent evt = (ConfigDataChangeEvent) event;
                    long dumpTs = evt.lastModifiedTs;
                    String dataId = evt.dataId;
                    String group = evt.group;
                    String tenant = evt.tenant;
                    String tag = evt.tag;
                    // 获取所有的成员
                    Collection<Member> ipList = memberManager.allMembers();

                    // In fact, any type of queue here can be
                    Queue<NotifySingleTask> queue = new LinkedList<NotifySingleTask>();
                    for (Member member : ipList) {
                        queue.add(new NotifySingleTask(dataId, group, tenant, tag, dumpTs, member.getAddress(), evt.isBeta));
                    }
                    ConfigExecutor.executeAsyncNotify(new AsyncTask(tpmAsyncRestTemplate, queue));
                }
            }

            @Override
            public Class<? extends Event> subscribeType() {
                return ConfigDataChangeEvent.class;
            }
        });
    }

    class AsyncTask implements Runnable {

        private Queue<NotifySingleTask> queue;

        private TpmAsyncRestTemplate restTemplate;

        public AsyncTask(TpmAsyncRestTemplate restTemplate, Queue<NotifySingleTask> queue) {
            this.restTemplate = restTemplate;
            this.queue = queue;
        }

        @Override
        public void run() {
            executeAsyncInvoke();
        }

        private void executeAsyncInvoke() {
            while (!queue.isEmpty()) {
                NotifySingleTask task = queue.poll();
                // 获取目标机器的地址
                String targetIp = task.getTargetIP();
                // 判断目标机器是否在集群中
                if (memberManager.hasMember(targetIp)) {
                    // 对目标机器进行健康检查
                    boolean unHealthNeedDelay = memberManager.isUnHealth(targetIp);
                    if (unHealthNeedDelay) {
                        log.info(">>> [😒 unhealth node] {}|{}|{}|{}|{}|{}|{}|{}|{}|{}|{}", InetUtils.getSelfIP(), task.getDataId(), task.getGroup(), task.getTenant(),
                                null, task.getLastModified(), InetUtils.getSelfIP(), "notify", NOTIFY_EVENT_UNHEALTH, 0, targetIp);
                        // 如果服务是非健康状态，重新提交任务到线程池当中
                        asyncTaskExecute(task);
                    } else {
                        // 如果该节点是健康的
                        Header header = Header.newInstance();
                        header.addParam(NOTIFY_HEADER_LAST_MODIFIED, String.valueOf(task.getLastModified()));
                        header.addParam(NOTIFY_HEADER_OP_HANDLE_IP, InetUtils.getSelfIP());
                        if (task.isBeta) {
                            header.addParam("isBeta", "true");
                        }
                        AuthHeaderUtil.addIdentityToHeader(header);
                        // http://ip:port/tpm/v1/cs/communication/dataChange?dataId={2}&group={3}&tenant={4}
                        restTemplate.get(task.url, header, Query.EMPTY, String.class, new AsyncNotifyCallBack(task));
                    }
                }
            }
        }
    }

    private void asyncTaskExecute(NotifySingleTask task) {
        // 计算任务的延时时间
        int delay = getDelayTime(task);
        Queue<NotifySingleTask> queue = new LinkedList<NotifySingleTask>();
        queue.add(task);
        // 重新封装 AsyncTask 任务
        AsyncTask asyncTask = new AsyncTask(tpmAsyncRestTemplate, queue);
        // 延时执行任务
        ConfigExecutor.scheduleAsyncNotify(asyncTask, delay, TimeUnit.MILLISECONDS);
    }

    class AsyncNotifyCallBack implements Callback<String> {

        private NotifySingleTask task;

        public AsyncNotifyCallBack(NotifySingleTask task) {
            this.task = task;
        }

        @Override
        public void onReceive(RestResult<String> result) {

            if (result.ok()) {
                log.info(">>> [😊 notify_event_ok] {}|{}|{}|{}|{}|{}|{}|{}|{}", InetUtils.getSelfIP(), task.getDataId(), task.getGroup(), task.getTenant(),
                        task.getLastModified(), InetUtils.getSelfIP(), "notify", 0, task.getTargetIP());
            } else {
                log.error(">>> [😒 notify-error] target:{} dataId:{} group:{} ts:{} code:{}", task.target, task.getDataId(),
                        task.getGroup(), task.getLastModified(), result.getCode());
                // 获取延迟时间并为任务设置失败计数，重新提交任务
                asyncTaskExecute(task);

                log.error(">>> [😒 notify-retry] target:{} dataId:{} group:{} ts:{}", task.target, task.getDataId(),
                        task.getGroup(), task.getLastModified());
            }
        }

        @Override
        public void onError(Throwable ex) {

            log.error(">>> [😒 notify-exception] target:{} dataId:{} group:{} ts:{} ex:{}", task.target, task.getDataId(),
                    task.getGroup(), task.getLastModified(), ex.toString());
            //get delay time and set fail count to the task
            asyncTaskExecute(task);
            log.error(">>> [😒 notify-retry] target:{} dataId:{} group:{} ts:{}", task.target, task.getDataId(),
                    task.getGroup(), task.getLastModified());
        }

        @Override
        public void onCancel() {

            log.error(">>> [😒 notify-exception] target:{} dataId:{} group:{} ts:{} method:{}", task.target,
                    task.getDataId(), task.getGroup(), task.getLastModified(), "CANCELED");

            //get delay time and set fail count to the task
            asyncTaskExecute(task);
            log.error(">>> [😒 notify-retry] target:{} dataId:{} group:{} ts:{}", task.target, task.getDataId(),
                    task.getGroup(), task.getLastModified());
        }
    }

    static class NotifySingleTask extends NotifyTask {

        private String target;

        public String url;

        private boolean isBeta;

        private static final String URL_PATTERN =
                "http://{0}{1}" + Constants.COMMUNICATION_CONTROLLER_PATH + "/dataChange" + "?dataId={2}&group={3}";

        // http://{0}{1}/v1/cs/communication/dataChange?dataId={2}&group={3}&tenant={4}
        private static final String URL_PATTERN_TENANT =
                "http://{0}{1}" + Constants.COMMUNICATION_CONTROLLER_PATH + "/dataChange"
                        + "?dataId={2}&group={3}&tenant={4}";

        // 当前任务失败的次数
        private int failCount;

        public NotifySingleTask(String dataId, String group, String tenant, long lastModified, String target) {
            this(dataId, group, tenant, lastModified, target, false);
        }

        public NotifySingleTask(String dataId, String group, String tenant, long lastModified, String target,
                                boolean isBeta) {
            this(dataId, group, tenant, null, lastModified, target, isBeta);
        }

        public NotifySingleTask(String dataId, String group, String tenant, String tag, long lastModified,
                                String target, boolean isBeta) {
            super(dataId, group, tenant, lastModified);
            this.target = target;
            this.isBeta = isBeta;
            try {
                dataId = URLEncoder.encode(dataId, Constants.ENCODE);
                group = URLEncoder.encode(group, Constants.ENCODE);
            } catch (UnsupportedEncodingException e) {
                log.error("URLEncoder encode error", e);
            }
            if (StringUtils.isBlank(tenant)) {
                this.url = MessageFormat.format(URL_PATTERN, target, EnvUtil.getContextPath(), dataId, group);
            } else {
                // http://ip:port/tpm/v1/cs/communication/dataChange?dataId={2}&group={3}&tenant={4}
                this.url = MessageFormat
                        .format(URL_PATTERN_TENANT, target, EnvUtil.getContextPath(), dataId, group, tenant);
            }
            if (StringUtils.isNotEmpty(tag)) {
                url = url + "&tag=" + tag;
            }
            failCount = 0;
        }

        @Override
        public void setFailCount(int count) {
            this.failCount = count;
        }

        @Override
        public int getFailCount() {
            return failCount;
        }

        public String getTargetIP() {
            return target;
        }

    }

    /**
     * 获取延迟时间并记录失败次数
     *
     * @param task 通知任务
     * @return 当前任务需要延时执行的时间
     */
    private static int getDelayTime(NotifySingleTask task) {
        // 获取当前任务失败的次数
        int failCount = task.getFailCount();
        // min = 500ms max = 500 + 7 * 7 * 1000 = 49500
        int delay = MIN_RETRY_INTERVAL + failCount * failCount * INCREASE_STEPS;
        if (failCount < MAX_COUNT) {
            // 失败次数自增1
            task.setFailCount(failCount + 1);
        }
        return delay;
    }

    /**
     * 最小重试间隔时间
     */
    private static final int MIN_RETRY_INTERVAL = 500;

    /**
     * 重试时间成长值
     */
    private static final int INCREASE_STEPS = 1000;

    /**
     * 最大失败次数的值
     */
    private static final int MAX_COUNT = 7;
}
